package cn.me.alphamq.broker.core.commitlog;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.core.consumequeue.ConsumeQueueMMapFileModel;
import cn.me.alphamq.common.model.broker.meta.MMapFileModel;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueIndexModel;
import cn.me.alphamq.common.util.FileNameUtil;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.util.AssertUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Optional;

/**
 * CommitLog 的 MMap 文件的模型
 *
 * @author f
 */
public class CommitLogMMapFileModel {

    private static final Logger log = LoggerFactory.getLogger(CommitLogMMapFileModel.class);

    /**
     * 底层用来读写数据的对象
     */
    private MMapFileModel mMapFileModel;

    /**
     * 本 MMap 文件对应的 topic 名称
     */
    private String topicName;

    /**
     * 指定 offset 做文件的映射
     *
     * @param topicName 消息的 topic
     */
    public void loadFileInMMap(String topicName) throws IOException {
        this.topicName = topicName;

        CommitLogTopicModel commitLogTopicModel = Container.fetchCommitLogTopicModelByTopic(topicName);
        AssertUtil.isNotNull(commitLogTopicModel, "topicName [" + topicName + "] is invalid");

        mMapFileModel = new MMapFileModel();
        mMapFileModel.mapFile(commitLogTopicModel.getCommitLogFileModel(), FileNameUtil.getCommitLogFileDir(topicName),
                BrokerConstant.COMMIT_LOG_MAX_SIZE);
    }

    /**
     * 从文件的指定 offset 开始读取内容
     *
     * @param readOffset
     * @param size
     * @return
     */
    public byte[] readContent(int readOffset, int size) {
        return mMapFileModel.read(readOffset, size);
    }

    /**
     * 更高性能的写入 api，异步刷盘
     *
     * @param message 指定的消息
     */
    public void writeContent(Message message) throws Exception {
        writeContent(message, false);
    }

    /**
     * 写入数据到磁盘当中，可以选择同步还是异步刷盘
     *
     * @param message 指定的消息
     * @param force   如果此参数为 false，则表示同步刷盘；否则是异步刷盘
     */
    public void writeContent(Message message, boolean force) throws Exception {
        byte[] writeContent = message.getBody();
        // 先加锁写入数据，并且获取数据在文件中的偏移量，然后再将索引数据分发给各个 ConsumeQueue
        int offset = mMapFileModel.write(writeContent, force);
        dispatch(message, offset);
    }

    /**
     * 将 CommitLog 的索引消息写入 ConsumerQueue 文件
     *
     * @param message   指定的消息
     * @param msgOffset 消息的偏移量
     */
    private void dispatch(Message message, int msgOffset) throws Exception {
        ConsumeQueueIndexModel consumeQueueIndexModel = new ConsumeQueueIndexModel();
        consumeQueueIndexModel.setCommitLogFileName(Integer.parseInt(mMapFileModel.getFileModel().getFileName()));
        consumeQueueIndexModel.setMsgOffset(msgOffset);
        consumeQueueIndexModel.setMsgLength(message.getBody().length);
        consumeQueueIndexModel.setRetryTimes(message.getCurRetryTimes());

        /*
            TODO 可以实现广播消费模式
            这里有两种场景：
            - 如果是广播消费模式，生产者只需要向某个队列投递消息
                就没有必要将索引数据放到队列中了，采用推模式直接将消息的元数据推送到消费者端
            - 如果是集群消费模式，生产者需要向每个订阅主题的消费者投递消息
                需要将索引数据放到队列中，采用拉模式等待消费者来拉取数据
         */
        int queueId = message.getQueueId();

        Optional<ConsumeQueueMMapFileModel> consumeQueueMMapFileModel = Container.getConsumeQueueMMapFileModelManager()
                .get(topicName).stream()
                .filter(item -> item.getQueueId().equals(queueId)).findFirst();
        if (consumeQueueMMapFileModel.isPresent()) {
            consumeQueueMMapFileModel.get().writeContent(consumeQueueIndexModel);
        } else {
            log.error("consumeQueueMMapFileModel is null, topicName = {}, queueId = {}", topicName, queueId);
        }
    }

    /**
     * 释放 MMap 内存占用
     */
    public void clean() {
        mMapFileModel.clean();
    }

}
